package ds_industry_2025.industry.gy_09.T3

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

import java.util.Properties
/*
    3、编写scala代码，使用Spark根据dws层的machine_produce_per_avgtime表,获取各设备生产耗时最长的两个产品的用时,将计算结果存
    入MySQL数据库shtd_industry的machine_produce_timetop2表中（表结构如下），然后在Linux的MySQL命令行中根据设备id降序排序，
    查询出前2条，将SQL语句复制粘贴至客户端桌面【Release\任务B提交结果.docx】中对应的任务序号下，将执行结果截图粘贴至客户端桌面
    【Release\任务B提交结果.docx】中对应的任务序号下；
 */
object t3 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t3")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    val conn=new Properties()
    conn.setProperty("user","root")
    conn.setProperty("password","123456")
    conn.setProperty("driver","com.mysql.jdbc.Driver")

    spark.table("dws.machine_produce_per_avgtime")
      .select("produce_machine_id","producetime")
      .createOrReplaceTempView("tmp")

    val r1 = spark.sql(
      """
        |select
        |machine_id,row,time
        |from(
        |select distinct
        |produce_machine_id as machine_id,
        |dense_rank() over(partition by produce_machine_id order by producetime desc) as row,
        |producetime as time
        |from tmp
        |) as r1
        |where row < 3
        |""".stripMargin)

    // todo 为什么这里使用first函数，是因为已经通过 filter(col("row") === 1) 过滤了数据，每个 machine_id 只会有一个 time 值。
    // todo 使用 first 函数可以提取该值。就把结果给到了新的列
    val t1=r1.filter(col("row")===1)
      .withColumn("row",lit("first_time"))
      .groupBy("machine_id")
      .pivot("row")
      .agg(first(col("time")))

    val t2 = r1.filter(col("row") === 2)
      .withColumn("row", lit("second_time"))
      .groupBy("machine_id")
      .pivot("row")
      .agg(first(col("time")))

    t1.join(t2,"machine_id")
      .write.mode("overwrite")
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_industry?useSSL=false","machine_produce_timetop2",conn)



    spark.close()
  }

}
